package master;

import java.io.BufferedOutputStream;
import java.io.ByteArrayOutputStream;
import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.File;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.io.StringReader;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.ListIterator;
import java.util.PriorityQueue;

import javax.xml.bind.JAXBContext;
import javax.xml.bind.JAXBException;
import javax.xml.bind.Marshaller;
import javax.xml.bind.Unmarshaller;
import javax.xml.parsers.ParserConfigurationException;
import javax.xml.transform.stream.StreamSource;

import org.w3c.dom.Document;
import org.w3c.dom.Element;
import org.xml.sax.SAXException;

import common.FileObject;
import common.JobObject;
import common.JobsPackage;
import common.LoginRRObject;
import common.Util;

public class ClientRequestThread extends Thread {

	MasterService cloudServerService;
	int revClientReqTCPPort; // port for receiving client request
	PriorityQueue<JobsPackage> jobsPackageQueue;
	
	public ClientRequestThread(MasterService cloudServerService, int revClientReqTCPPort, PriorityQueue<JobsPackage> jobsPackageQueue) {
		this.cloudServerService = cloudServerService;
		this.revClientReqTCPPort = revClientReqTCPPort;
		this.jobsPackageQueue = jobsPackageQueue;
	}
	
	public void run() {
		try{
			//			SSLServerSocketFactory sslserversocketfactory = (SSLServerSocketFactory) SSLServerSocketFactory.getDefault();
			//			SSLServerSocket listenSocket = (SSLServerSocket) sslserversocketfactory.createServerSocket(managerTCPPort);
			ServerSocket listenSocket = new ServerSocket(revClientReqTCPPort);

			while (true) {
				Socket clientSocket = listenSocket.accept();
				System.out.println("SYSINFO - Accept Client TCP connection.");
				
				DoRequest doRequest = new DoRequest(clientSocket, cloudServerService, jobsPackageQueue);
				doRequest.start();

			}
		}catch(IOException e){System.out.println("Listen:"+e.getMessage());}
	}
}

class DoRequest extends Thread {
	private Socket clientSocket;
	MasterService cloudServerService;
	PriorityQueue<JobsPackage> jobsPackageQueue;
	static final String WAITING = "WAITING";
	
	public DoRequest(Socket clientSocket, MasterService cloudServerService, PriorityQueue<JobsPackage> jobsPackageQueue) {
		this.clientSocket = clientSocket;
		this.cloudServerService = cloudServerService;
		this.jobsPackageQueue = jobsPackageQueue;
	}
	
	public void run() {
		try {
			InputStream in = clientSocket.getInputStream();
			DataInputStream dataIn = new DataInputStream(in);
	
			OutputStream out = clientSocket.getOutputStream();
			DataOutputStream dataOut = new DataOutputStream(out);
			
			String request = dataIn.readUTF();
			System.out.println("SYSINFO - Received Client request.");
			System.out.println("DEBUG - xml recevied: " + request);
			
			Document inDoc = Util.stringToDoc(request);
			Element inRoot = inDoc.getDocumentElement();
			
			String reply = null;
			if (inRoot.getAttribute("type").equals("Login")) {
				System.out.println("DEBUG - It is a login request.");
				//reply = doLogin(inDoc, clientSocket);
				reply = doLogin(request, dataOut, clientSocket);
			} 
			else if (inRoot.getAttribute("type").equals("Register")) {
				System.out.println("DEBUG - It is a register request.");
				reply = doRegister(inDoc);
			} 
			else if (inRoot.getAttribute("type").equals("SendJobs")) {
				//SendJobs -- start to schedule immediately
				System.out.println("DEBUG - It is a SendJobs request.");
				//reply = doSendJobs(inDoc);
				reply = doSendJobs(request,dataIn);
			} 
			else if (inRoot.getAttribute("type").equals("SendFiles")) {
				System.out.println("DEBUG - It is a SendFiles request.");
				reply = doSendFiles(inDoc);
			} 
			else if (inRoot.getAttribute("type").equals("RetrieveOutput")) {
				System.out.println("DEBUG - It is a RetrieveOutput request.");
				reply = doRetrieveOutput(inDoc, dataOut);
			} 
			else if (inRoot.getAttribute("type").equals("QueryJobs")) {
				System.out.println("DEBUG - It is a QueryJobs request.");
				reply = doQueryJobs(inDoc);
			} else if (inRoot.getAttribute("type").equals("KillJob")) {
				System.out.println("DEBUG - It is a KillJobs request.");
				reply = doKillJob(inDoc);
			} else {
				System.out.println("SYSINFO - It is NOT a valid request. No action will be performed.");
			}
			
			if (reply!=null) dataOut.writeUTF(reply);
			System.out.println("SYSINFO - Send reply.");
			dataOut.close();
			
		} catch(IOException e){
			System.out.println("Conection:"+e.getMessage());
		} catch (ParserConfigurationException e) {
			e.printStackTrace();
		} catch (SAXException e) {
			e.printStackTrace();
		} finally{
			try{
				if(clientSocket != null) clientSocket.close();
			}
			catch(IOException e){}
		}
	}

	private String doQueryJobs(Document inDoc) {
		Element inRoot = inDoc.getDocumentElement();
		String clientID = inRoot.getElementsByTagName("clientID").item(0).getTextContent();
		System.out.println("SYSINFO - Query Jobs by " + clientID);
		JobsPackage jp = cloudServerService.jobPackageInfo.get(clientID);
		if (jp==null) {
			jp = new JobsPackage();
			jp.setBudget(0);
			jp.setClientID(clientID);
			jp.setCost(0);
			jp.setDeadline((long)0);
			jp.setFrom("Master");
			jp.setJobs(new ArrayList());
			jp.setJpCompletedIndcator(true);
			jp.setTime((long)0);
			jp.setTimestamp((long)0);
			jp.setStartTime((long)0);
			jp.setEndTime((long)0);
			jp.setType(null);
		}
		
		if (jp.getNumberOfJobsCompleted() < jp.getNumberOfJobs())
			jp.setEndTime(cloudServerService.getTime());

		JAXBContext context;
		try {
			context = JAXBContext.newInstance(JobsPackage.class);
			Marshaller m = context.createMarshaller();
		    m.setProperty(Marshaller.JAXB_FORMATTED_OUTPUT, true);
		    ByteArrayOutputStream baos = new ByteArrayOutputStream();
		    m.marshal(jp, baos);
		    String clientQueryXML = baos.toString();
			return clientQueryXML;
		} catch (JAXBException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		/*JAXBContext context;
		try {
			context = JAXBContext.newInstance(QueryObject.class);
			Marshaller m = context.createMarshaller();
		    m.setProperty(Marshaller.JAXB_FORMATTED_OUTPUT, true);
		    QueryObject queryObject = new QueryObject();
		    queryObject.setNumberOfJobs(jp.getNumberOfJobs());
		    queryObject.setNumberOfJobsCompleted(jp.getNumberOfJobsCompleted());
		    queryObject.setNumberOfJobsRunning(jp.getNumberOfJobsRunning());
		    queryObject.setNumberOfJobsWaiting(jp.getNumberOfJobsWaiting());
		    ByteArrayOutputStream baos = new ByteArrayOutputStream();
		    m.marshal(queryObject, baos);
		    String clientObjectXML = baos.toString();
			return clientObjectXML;
		} catch (JAXBException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}*/
		return null;
	}

    private String doKillJob(Document inDoc){
        //Element inRoot = inDoc.getDocumentElement();
		String clientID = inDoc.getElementsByTagName("clientID").item(0).getTextContent();
        int jobID = Integer.parseInt(inDoc.getElementsByTagName("jobID").item(0).getTextContent());
        if(cloudServerService.killJobInWorkerJobList(clientID, jobID)){
            System.out.println("SYSINFO - Kill job complete");
            return "KILLED";
        }else{
            System.out.println("SYSINFO - Can not find the job in both Master and Workers");
        }
        return null;
    }

	private String doRetrieveOutput(Document inDoc, DataOutputStream dataOut) {
		Element inRoot = inDoc.getDocumentElement();
		String clientID = inRoot.getElementsByTagName("clientID").item(0).getTextContent();
		System.out.println("SYSINFO - Retrive Results by " + clientID);
		cloudServerService.sendBackResult(clientID, dataOut);
		return null;
	}

	private String doSendFiles(Document inDoc) {
		// TODO Spawn Thread to recevie Files
		return null;
	}

	private String doSendJobs(String request, DataInputStream dataIn) {

		try {
			/********** JobPackage Information Transfer **********/
			JAXBContext context = JAXBContext.newInstance(JobsPackage.class);
			Unmarshaller u = context.createUnmarshaller();
			StringBuffer xmlStr = new StringBuffer(request);
			JobsPackage jp = (JobsPackage) u.unmarshal(new StreamSource(new StringReader(xmlStr.toString())));

			// ensure the Status of each Job inside the JobPackage transfered from Client is WAITING
			for (int count = 0; count < jp.getJobs().size(); count++) {
				jp.getJobs().get(count).setStatus(WAITING);
			}
			jp.setStartTime(cloudServerService.getTime());
			cloudServerService.queueJobsPackage(jp);
			cloudServerService.jobPackageInfo.put(jp.getClientID(), jp);

			ListIterator<JobObject> iter = jp.getJobs().listIterator();
			List<FileObject> fileList = new ArrayList<FileObject>();
			while (iter.hasNext()) {

				Long fileSize = dataIn.readLong();
				JobObject job = iter.next();
				System.out.println("DEBUG - clientID = " + jp.getClientID());
				System.out.println("DEBUG - fileName = " + job.getExecuteFileName());
				System.out.println("DEBUG - parameters = "	+ job.getParameters());
				System.out.println("DEBUG - time = " + job.getTime());
				System.out.println("DEBUG - jobID = " + job.getJobID());
				System.out.println("DEBUG - priority = " + job.getPriority());
				System.out.println("DEBUG - fileSize = " + fileSize);
				String filePath = "masterJobs"; // default jobs folder (but can be assign to another one)

				String savePath = filePath + "/" + jp.getClientID();
				job.setExecuteFilePath(filePath);
				File file = new File(savePath);

				// if file path not exist, create a directory for it
				if(file.exists()==false){
					file.mkdirs();
				}
				System.out.println("DEBUG - Created directory: " + file);

				String saveFile = savePath + "/" + job.getExecuteFileName() + ".class"; // NTD: add jobID to fileName
				FileObject fileObject = new FileObject(saveFile, fileSize);
				fileList.add(fileObject);
			}

			/********** File Transfer **********/
			int bufferSize = 8192;
			byte[] buf = new byte[bufferSize];
			
			Iterator fileIter = fileList.iterator();
			FileObject fileObject = new FileObject();
			
			int length = 0;
			while ((length = dataIn.read(buf)) != -1) {
				System.out.println("DEBUG - Read data from connection into buffer.");

				int bufRead = 0;

				while (bufRead != bufferSize && fileIter.hasNext()) {
					System.out.println("DEBUG - Read data from buffer.");
					if (fileObject.isCompleted() && fileIter.hasNext()){
						System.out.println("DEBUG - A new executeFile.");
						fileObject = (FileObject) fileIter.next();
					}
					
					if (fileObject.getFileSize() == 0) {
						System.out.println("DEBUG - File size is 0.");
						System.out.println("SYSINFO - File saved at " + fileObject.getFileName());
						DataOutputStream fileOut = new DataOutputStream(new BufferedOutputStream(new FileOutputStream(fileObject.getFileName())));
						fileOut.write(buf, bufRead, (int) fileObject.getFileSize());
						fileOut.close();
						fileObject.setCompleted(true);
					}
					
					long befLeftLong = fileObject.getFileSize() - fileObject.getFileBufSave();
					int befLeft = (int) befLeftLong; // not safe casting
					
					if (bufferSize - bufRead >= befLeft && befLeft != 0) {
						System.out.println("DEBUG - Completed in one write operation. " + fileObject.getFileName());
						DataOutputStream fileOut = new DataOutputStream(new BufferedOutputStream(new FileOutputStream(fileObject.getFileName())));
						fileOut.write(buf, bufRead, (int) fileObject.getFileSize());
						fileOut.close();
						System.out.println("SYSINFO - File saved at " + fileObject.getFileName());
						fileObject.setFileBufSave(fileObject.getFileSize());
						bufRead += fileObject.getFileSize();
						fileObject.setCompleted(true);
						
					} else if (bufferSize - bufRead < befLeft) { // NTD: handle
						System.out.println("DEBUG - Completed in multiple write operation. " + fileObject.getFileName());
						DataOutputStream fileOut = new DataOutputStream(new BufferedOutputStream(new FileOutputStream(fileObject.getFileName())));
						fileOut.write(buf, bufRead, bufferSize);
						System.out.println("DEBUG - File transfer haven't completed");
						fileObject.setFileBufSave(bufferSize - bufRead);
						fileOut.close(); // file has been written out, Master may scheduler an uncompleted file
						bufRead = bufferSize;
						fileObject.setCompleted(false);
						
					}
				}
			}
		} catch (JAXBException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (FileNotFoundException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}

		return null;
	}

	private String doRegister(Document inDoc) {
		// TODO Spwan Thread to Register user
		return null;
	}

	private String doLogin(String request, DataOutputStream dataOut, Socket clientSocket) {
		JAXBContext context;
		try {
			
			context = JAXBContext.newInstance(LoginRRObject.class);
			Unmarshaller u = context.createUnmarshaller();
			StringBuffer xmlStr = new StringBuffer(request);
			LoginRRObject loginRRObject = (LoginRRObject) u.unmarshal(new StreamSource(new StringReader(xmlStr.toString())));
			
			Marshaller m = context.createMarshaller();
		    m.setProperty(Marshaller.JAXB_FORMATTED_OUTPUT, true);
		    
		    loginRRObject.getClient().setClientID("CS" + cloudServerService.getClientNumber());
		    cloudServerService.clientHashtable.put(loginRRObject.getClient().getClientID(), loginRRObject.getClient());
		    cloudServerService.incrementClientNumber();
		    loginRRObject.getClient().setClientPort(clientSocket.getPort());
		    
		    ByteArrayOutputStream baos = new ByteArrayOutputStream();
		    m.marshal(loginRRObject, baos);
		    String clientXML = baos.toString();
		    
			return clientXML;
		} catch (JAXBException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		return null;
	    
	}
}
